require 'kafka'
require 'singleton'

module KafkaConstant
  DEFAULT_TOPIC = 'test'
  CONSUMER_GROUP = 'my-consumer'
  KAFKA_CLIENTID = 'my-application'

  KAFKA_HOSTS = ['localhost:9092'].freeze

  DEFAULT_RECEIVE_BLOCK = ->(message) {
    puts "[#{message.topic}], [#{message.partition}]"
    puts "[#{message.offset}], [#{message.key}], [#{message.value}]"
  }
end

class KafkaClient
  include Singleton
  attr_reader :kafka

  class << self
    def send_message(messages)
      instance.send_message(messages)
    end

    def receive_message(&block)
      instance.receive_message(&block)
    end

    def close
      instance.close
    end
  end

  def initialize
    @kafka = Kafka.new(KafkaConstant::KAFKA_HOSTS, client_id: KafkaConstant::KAFKA_CLIENTID)
  end

  def send_message(messages)
    producer.produce(messages, topic: KafkaConstant::DEFAULT_TOPIC)
    producer.deliver_messages
  end

  def receive_message(&block)
    block ||= KafkaConstant::DEFAULT_RECEIVE_BLOCK
    consumer.each_message {|message| block.call message}
  end

  def close
    close_producer
    close_consumer
  end

  private

  def producer
    @producer ||= @kafka.async_producer(delivery_interval: 1)
    # @producer ||= @kafka.producer
  end

  def consumer
    if @consumer.nil?
      @consumer = kafka.consumer(group_id: KafkaConstant::CONSUMER_GROUP)
      @consumer.subscribe(KafkaConstant::DEFAULT_TOPIC)
    end
    @consumer
  end

  def close_producer
    if @producer
      @producer.shutdown
      @producer = nil
    end
  end

  def close_consumer
    if @consumer
      @consumer.stop
      @consumer = nil
    end
  end
end

at_exit {
  KafkaClient.close
}
